package net.flyinggroup.scst.mimo;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import java.util.function.Function;

@SpringBootApplication
public class SampleApplication {
    public static void main(String[] args) {
        SpringApplication.run(SampleApplication.class, args);
    }

    @Bean
    public Function<Flux<Integer>, Tuple2<Flux<String>, Flux<String>>> scatter() {
        return flux -> {
            Flux<Integer> connectedFlux = flux.publish().autoConnect(2);
            Sinks.Many<String> even = Sinks.many().unicast().onBackpressureBuffer();
            Sinks.Many<String> odd = Sinks.many().unicast().onBackpressureBuffer();
            Flux<Integer> evenFlux = connectedFlux.filter(number -> number % 2 == 0).doOnNext(number -> even.tryEmitNext("EVEN:" + number));
            Flux<Integer> oddFlux = connectedFlux.filter(number -> number % 2 != 0).doOnNext(number -> odd.tryEmitNext("ODD:" + number));
            return Tuples.of(even.asFlux().doOnSubscribe(x -> evenFlux.subscribe()), odd.asFlux().doOnSubscribe(x -> oddFlux.subscribe()));
        };
    }
}
